package org.example.music;

import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * MusicReducer类继承自TableReducer，用于处理音乐播放次数的归约任务
 * 它的作用是将具有相同key（例如音乐ID）的所有播放次数相加，然后将结果插入到HBase表中
 */
public class MusicReducer extends TableReducer<Text, IntWritable, Text> {
    /**
     * reduce方法是MusicReducer类的核心方法，它接收一个key和与之相关联的所有value进行处理
     *
     * @param key     当前处理的键，例如音乐ID
     * @param values  与key相关联的所有播放次数
     * @param context 上下文对象，用于与Hadoop框架交互，例如写入结果
     * @throws IOException          当写入输出发生错误时抛出
     * @throws InterruptedException 当任务被中断时抛出
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, Mutation>.Context context) throws IOException, InterruptedException {
        //播放次数
        int playCount = 0;
        // 遍历所有value，计算总播放次数
        for (IntWritable value : values) {
            playCount += value.get();
        }

        // 创建一个Put对象，用于将数据插入HBase表中
        Put put = new Put(Bytes.toBytes(key.toString()));
        // 指定列和值，这里将播放次数作为列的值插入到HBase表中
        put.addColumn(Bytes.toBytes("details"), Bytes.toBytes("rank"), Bytes.toBytes(playCount));
        // 将键和Put对象写入到HBase表中
        context.write(key, put);
    }
}

